package cn.ai.boot.config;

import com.google.common.collect.Lists;
import org.apache.commons.lang3.ArrayUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.util.StopWatch;

import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;

/**
 * @author jie.zhao@dms365.com
 * @Date 2025/7/1 14:14
 */
public class DocConsumerService {
    private static final Logger LOGGER = LoggerFactory.getLogger(DocConsumerService.class);

    public static void main(String[] args) {
        List<DocFolderSaveDTO> addFolderList
                = new ArrayList<>();

        ExecutorService executorService = Executors.newFixedThreadPool(50);

        for (int i = 0; i < 510; i++) {
            addFolderList.add(new DocFolderSaveDTO(UUID.randomUUID().toString()));
        }
        try {
            StopWatch stopWatch = new StopWatch();
            stopWatch.start("开始");
            List<Map<String, Object>> maps = saveFolderAsync4(executorService, addFolderList, "", 5);
            stopWatch.stop();
            System.out.println(stopWatch.getTotalTimeMillis());
            System.out.println(maps.size());

        } catch (Exception e) {
            LOGGER.info("saveFolderAsync.Exception", e);
        } finally {
            executorService.shutdown();
        }
        System.out.println("结束");

    }


    public static List<Map<String, Object>> saveFolderAsync4(ExecutorService executorService, List<DocFolderSaveDTO> addFolderList, String docCode, int maxPermits
    ) {
        List<List<DocFolderSaveDTO>> partition = Lists.partition(addFolderList, 20);
        AsyncServiceUtils asyncService = new AsyncServiceUtils();
        List<Map<String, Object>> allMap = Collections.synchronizedList(new ArrayList<>());

        List<List<Map<String, Object>>> saveFolderAsync = asyncService.executeConcurrentTaskWithResult(executorService, partition, maxPermits,
                list -> {
                    List<Map<String, Object>> data = new ArrayList<>();
                    for (DocFolderSaveDTO docFolderSaveDTO : list) {
                        Map<String, Object> map = new HashMap<>();
                        map.put("1", docFolderSaveDTO.getName());
                        data.add(map);
                    }
                    return data;
                }, "saveFolderAsync");

        for (List<Map<String, Object>> maps : saveFolderAsync) {

            allMap.addAll(maps);
        }
        return allMap;
    }

    public static List<Map<String, Object>> saveFolderAsync2(ExecutorService executorService, List<DocFolderSaveDTO> addFolderList, String docCode, int maxPermits
    ) {
        List<List<DocFolderSaveDTO>> partition = Lists.partition(addFolderList, 20);

        // 设置线程池
        Semaphore semaphore = new Semaphore(maxPermits); // 控制最大并发数量
        AtomicBoolean isException = new AtomicBoolean(false);
        List<Map<String, Object>> allMap = Collections.synchronizedList(new ArrayList<>());
        List<Future<List<Map<String, Object>>>> futures = new CopyOnWriteArrayList<>();

        for (List<DocFolderSaveDTO> docFolderSaveDTOS : partition) {
            if (isException.get()) {
                break;
            }
            try {
                semaphore.acquire();
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            futures.add(executorService.submit(() -> {
                try {
                    List<Map<String, Object>> data = saveFolder(docFolderSaveDTOS);
                    LOGGER.info("executorService：{}", allMap.size());

                    return data;
                } catch (Exception e) {
                    isException.set(true);
                    throw new RuntimeException("线程执行失败");
                } finally {
                    semaphore.release();
                }
            }));

        }
        for (Future<List<Map<String, Object>>> future : futures) {
            try {
                List<Map<String, Object>> maps = future.get();
                LOGGER.info("开始获取结果：{}", maps.size());
                allMap.addAll(maps);
            } catch (Exception e) {
                throw new RuntimeException("线程执行失败");
            }
        }
        if (isException.get()) {
            throw new RuntimeException("获取文件夹失败");
        }

        return allMap;
    }


    public static List<Map<String, Object>> saveFolderAsync(ExecutorService executorService, List<DocFolderSaveDTO> addFolderList, String docCode, int maxPermits
    ) {
        List<List<DocFolderSaveDTO>> partition = Lists.partition(addFolderList, 20);

        // 设置线程池
        Semaphore semaphore = new Semaphore(maxPermits); // 控制最大并发数量
        CyclicBarrier cyclicBarrier = new CyclicBarrier(partition.size() + 1);
        AtomicBoolean isException = new AtomicBoolean(false);
        List<Map<String, Object>> allMap = Collections.synchronizedList(new ArrayList<>());
        for (List<DocFolderSaveDTO> docFolderSaveDTOS : partition) {
            try {
                if (isException.get()) {
                    break;
                }
                semaphore.acquire();
                executorService.submit(() -> {
                    try {
                        List<Map<String, Object>> data = saveFolder(docFolderSaveDTOS);
                        allMap.addAll(data);
                        if (allMap.size() > 500) {
//                            System.out.println(1 / 0);
                        }
                    } catch (Exception e) {
                        isException.set(true);
                    } finally {
                        semaphore.release();
                        try {
                            cyclicBarrier.await();
                        } catch (InterruptedException | BrokenBarrierException e) {
                            Thread.currentThread().interrupt();
                        }
                    }
                });
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            } catch (Exception e) {
                LOGGER.debug("saveFolderAsync.Exception", e);
            }
        }
        try {
            cyclicBarrier.await();
        } catch (InterruptedException | BrokenBarrierException e) {
            throw new RuntimeException(e);
        }
        if (isException.get()) {
            throw new RuntimeException("获取文件夹失败");
        }

        return allMap;
    }

    private static AtomicInteger atomicInteger = new AtomicInteger();

    public static List<Map<String, Object>> saveFolder(List<DocFolderSaveDTO> batchSaveFolderDTO) throws InterruptedException {

        LOGGER.info("处理第批,数据：{}", batchSaveFolderDTO.size());
        Thread.sleep(2000L);
        int v = atomicInteger.addAndGet(batchSaveFolderDTO.size());

        if (v > 200) {
            System.out.println(1 / 0);
        }
        return batchSaveFolderDTO.stream()
                .map(dto -> {
                    Map<String, Object> map = new HashMap<>();
                    map.put("name", dto.getName());
                    return map;
                })
                .collect(Collectors.toList());
    }

}
